package com.wl.demosparkstream

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object RDDQueueDStream {


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(5))

    var rddQueue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]()
    val queueStream = ssc.queueStream(rddQueue)
    val value = queueStream.map(r => (r % 10, 1)).reduceByKey(_ + _)
    value.print()

    ssc.start()
    for (i <- 1 to 5){
      rddQueue.synchronized {
        val range = (1 to 100).map(_*i)
        rddQueue += ssc.sparkContext.makeRDD(range, 2)
      }
      Thread.sleep(2000)
    }

    ssc.stop()

  }
}
